package com.gitee.dengmin.logback;

import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.kafka.clients.producer.KafkaProducer;

/**
 * @Author dengmin
 * @Created 2020/6/8 下午5:01
 */
public class KafkaProducePool {
    private static final long serialVersionUID = 1L;
    private GenericObjectPool<KafkaProducer<String,String>> pool;

    public KafkaProducePool(String hosts){
        GenericObjectPoolConfig<KafkaProducer<String,String>> poolConfig = new GenericObjectPoolConfig<KafkaProducer<String,String>>();
        poolConfig.setMinIdle(1);
        poolConfig.setMaxIdle(5);
        poolConfig.setMaxTotal(10);
        poolConfig.setMaxWaitMillis(1000);
        pool = new GenericObjectPool<KafkaProducer<String, String>>(new KafkaProducerFactory(hosts),poolConfig);
    }

    public KafkaProducer<String,String> get() throws Exception {
        return pool.borrowObject();
    }

    public void returnObject(KafkaProducer<String,String> producer){
        pool.returnObject(producer);
    }

}
